<?php

namespace common\service;

use support\bootstrap\Redis;
use support\SingletonTrait;

class TenantEventService
{
    use SingletonTrait;

    const MAX_RETRY_TIME = 3;
    const MESSAGE_EXPIRE_TIME = 3600 * 24 * 7;
    const MESSAGE_OVERDUE_EXECUTE_TIME = 60 * 5;

    /**
     * 消息过期时间
     *
     * @return int
     */
    protected function getMessageExpireTime()
    {
        return OptionsService::getInstance()->getOptionsConfigItemData('tenant_event_config', 'expire_time', self::MESSAGE_EXPIRE_TIME);
    }

    /**
     * 消息执行过期时间
     *
     * @return int
     */
    protected function getMessageOverdueExecuteTime()
    {
        return OptionsService::getInstance()->getOptionsConfigItemData('tenant_event_config', 'overdue_execute_time', self::MESSAGE_OVERDUE_EXECUTE_TIME);
    }

    /**
     * 租户事件缓存key
     * @param int $tenantId
     * @return string
     */
    protected function getTenantEventQueueKey(int $tenantId)
    {
        return 'tenant_event_log_' . $tenantId;
    }

    /**
     * @return Redis
     */
    protected function getRedis()
    {
        return Redis::class;
    }

    /**
     * 清空过期事件
     * @param $tenantId
     */
    public function clearExpireMessage($tenantId)
    {
        $messageIdList = $this->getRedis()::zRangeByScore($this->getTenantEventQueueKey($tenantId) . ':queue_set', 0, time() - $this->getMessageExpireTime());
        foreach ($messageIdList as $messageId) {
            $this->remove($tenantId, $messageId);
        }
    }

    /**
     * 超时队列事件重新加入队尾
     * @param $tenantId
     */
    public function overdueMessageReQueue($tenantId)
    {
        $messageIdList = $this->getRedis()::zRangeByScore(
            $this->getTenantEventQueueKey($tenantId) . ':queue_inprogress',
            time() - $this->getMessageExpireTime(),
            time() - $this->getMessageOverdueExecuteTime()
        );
        foreach ($messageIdList as $messageId) {
            // 重新排队
            $this->getRedis()::lPush($this->getTenantEventQueueKey($tenantId) . ":queue", $messageId);
            $this->getRedis()::zRem($this->getTenantEventQueueKey($tenantId) . ":queue_inprogress", $messageId);
        }
    }

    /**
     * 发布一个事件
     * @param int $tenantId
     * @param $eventKey
     * @param null $eventData
     * @param null $messageId
     * @throws \Exception
     */
    public function push(int $tenantId, $eventKey, $eventData = null, $messageId = null)
    {
        $this->clearExpireMessage($tenantId);
        $this->overdueMessageReQueue($tenantId);
        $messageId = empty($messageId) ? create_uuid() : $messageId;
        $data = [
            'event_key' => $eventKey,
            'event_data' => $eventData,
        ];
        $this->pushData($tenantId, $data, $messageId);
    }

    /**
     * 发布消息
     * @param $tenantId
     * @param $data
     * @param null $messageId
     * @throws \Exception
     */
    private function pushData($tenantId, $data, $messageId = null)
    {
        $messageId = empty($messageId) ? create_uuid() : $messageId;
        // 保证消息队列里没有重复的消息在排队
        if ($this->getRedis()::zAdd($this->getTenantEventQueueKey($tenantId) . ':queue_set', time(), $messageId) == 1) {
            // 排队
            $this->getRedis()::lPush($this->getTenantEventQueueKey($tenantId) . ":queue", $messageId);
            // 写入 data
            $this->getRedis()::hSet($this->getTenantEventQueueKey($tenantId) . ":message_data", $messageId, json_encode($data));
        }
    }

    /**
     * 获取消息
     * @param int $tenantId
     * @return array|null
     */
    private function pullData(int $tenantId)
    {
        // 弹出右侧队首
        $k = $this->getRedis()::rPop($this->getTenantEventQueueKey($tenantId) . ":queue");
        if (empty($k)) {
            return null;
        }
        // 取出event_data
        $v = $this->getRedis()::hGet($this->getTenantEventQueueKey($tenantId) . ":message_data", $k);
        $this->getRedis()::zAdd($this->getTenantEventQueueKey($tenantId) . ':queue_inprogress', time(), $k);
        return [$k, $v];
    }

    /**
     * 取出一条消息
     * @param int $tenantId
     * @return array|null
     */
    public function pop(int $tenantId)
    {
        $this->clearExpireMessage($tenantId);
        $this->overdueMessageReQueue($tenantId);
        $data = $this->pullData($tenantId);
        if (empty($data)) {
            return null;
        }
        list($messageId, $messageData) = $data;
        return [
            'message_id' => $messageId,
            'message_data' => json_decode($messageData, true),
        ];
    }

    /**
     * 发布检查
     * @param $tenantId
     * @param $messageId
     * @return bool
     */
    private function pushCheck($tenantId, $messageId)
    {
        $k = $this->getTenantEventQueueKey($tenantId) . ':retry_check:' . $messageId;
        $old = $this->getRedis()::incr($k, 1);
        $this->getRedis()::expire($k, 3600 * 24);
        if (empty($old)) {
            return true;
        }

        return $old < self::MAX_RETRY_TIME;
    }

    /**
     * 重新发布消息
     * @param null $messageData
     * @throws \Exception
     */
    public function reAdd($tenantId, $messageId, $messageData = null)
    {
        if (!$this->pushCheck($tenantId, $messageId)) {
            return false;
        }
        $this->pushData($tenantId, $messageData, $messageId);
        return true;
    }

    /**
     * 标记成功
     *
     * @param int $tenantId
     * @param string $messageId
     * @return void
     */
    public function ack($tenantId, $messageId)
    {
        $this->clearExpireMessage($tenantId);
        $this->overdueMessageReQueue($tenantId);
        return $this->remove($tenantId, $messageId);
    }

    /**
     * 移除一条消息
     * @param $tenantId
     * @param $messageId
     */
    private function remove($tenantId, $messageId)
    {
        // todo 优化为一条命令执行 成功都成功 失败都失败
        // 队列集合删除
        $this->getRedis()::zRem($this->getTenantEventQueueKey($tenantId) . ':queue_set', $messageId);
        // 队列list删除
        $this->getRedis()::lRem($this->getTenantEventQueueKey($tenantId) . ':queue', 0, $messageId);
        // 队列inprogress zset删除
        $this->getRedis()::zRem($this->getTenantEventQueueKey($tenantId) . ':queue_inprogress', 0, $messageId);
        // 队列data删除
        $this->getRedis()::hDel($this->getTenantEventQueueKey($tenantId) . ":message_data", $messageId);
    }

}
